package com.example.demo.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.MimeTypeUtils;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import com.example.common.streams.KafkaStreams;

import reactor.core.publisher.Mono;

@RestController
@RequestMapping("/test")
public class TestController {

	@Autowired
	private KafkaStreams kafkaStreams;

	@GetMapping("{id}")
	public Mono<Long> findByBatchNumberPage(@PathVariable("id") Long id) {
		System.out.println(id);
		try {
			var messageChannel = kafkaStreams.outboundLogStreams();

			int i = 1;
			while (i < 10) {
				messageChannel.send(MessageBuilder.withPayload(i)
						.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
						.setHeader("partitionKey", "service-error-logs").build());
				i++;
			}

			int j = 9990;
			while (j < 10000) {
				messageChannel.send(MessageBuilder.withPayload(j)
						.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
						.setHeader(KafkaHeaders.TOPIC, "test0811").setHeader("partitionKey", "service-error-logs")
//						.setHeaders(new MessageHeaderAccessor())
						.build());
				j++;
			}
		} catch (Exception ex) {
			ex.printStackTrace();
		}
		return Mono.justOrEmpty(id);
	}

}
